package com.zheng.api.server.jms;

import com.zheng.api.server.service.mq.um.Mq4UmbrellaMachineNotify;
import com.zheng.api.server.service.mq.um.Mq4UmbrellaWechatappNotify;
import com.zheng.common.constant.MqConstant;
import org.apache.activemq.command.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.jms.*;
import javax.jms.Message;

/**
 * MQ消费者
 * Created by shuzheng on 2017/2/19.
 */
public class DefaultMessageQueueListener implements MessageListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultMessageQueueListener.class);

    @Autowired
    ThreadPoolTaskExecutor threadPoolTaskExecutor;
    @Autowired
    Mq4UmbrellaWechatappNotify mq4UmbrellaWechatappNotify;
    @Autowired
    Mq4UmbrellaMachineNotify mq4UmbrellaMachineNotify;

    @Override
    public void onMessage(final Message message) {
        // 使用线程池多线程处理
        threadPoolTaskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                //JMS规范中的消息类型包括TextMessage、MapMessage、ObjectMessage、BytesMessage、和StreamMessage等五种
                //如果是text消息
                if (message instanceof ActiveMQTextMessage) {
                    ActiveMQTextMessage textMessage = (ActiveMQTextMessage) message;
                    String msg = "";
                    try {
                        LOGGER.info("消费：{}", textMessage.getText());
                        //topic://bjs.server.topic.umbrella.wechatapp
                        String destination = textMessage.getJMSDestination().toString().replaceAll("topic://","");
                        msg = textMessage.getText();
                        if(MqConstant.ZHENG_SERVER_TOPIC_UMBRELLA_WECHATAPP_NAME.equals(destination)){
                            mq4UmbrellaWechatappNotify.receive(msg);
                        }else if(MqConstant.ZHENG_SERVER_TOPIC_UMBRELLA_MACHINE_NAME.equals(destination)){
                            mq4UmbrellaMachineNotify.receive(msg);
                        }else {
                            LOGGER.error("未找到消费者：{}", msg);
                        }
                    } catch (Exception e) {
                        LOGGER.error("消费：{}", msg,e);
                    }
                }
                //如果是Bytes消息
                else if(message instanceof ActiveMQBytesMessage){
                    ActiveMQBytesMessage bytesMessage = (ActiveMQBytesMessage) message;
                    String msg = "";
                    try {
                        byte[] byteContent = new byte[1024];
                        int length = -1;
                        StringBuffer content = new StringBuffer();
                        while ((length = bytesMessage.readBytes(byteContent)) != -1) {
                            content.append(new String(byteContent, 0, length));
                        }
                        LOGGER.info("消费：{}", content.toString());
                        //topic://bjs.server.topic.umbrella.wechatapp
                        String destination = message.getJMSDestination().toString().replaceAll("topic://","");
                        msg = content.toString();
                        if(MqConstant.ZHENG_SERVER_TOPIC_UMBRELLA_WECHATAPP_NAME.equals(destination)){
                            mq4UmbrellaWechatappNotify.receive(msg);
                        }else if(MqConstant.ZHENG_SERVER_TOPIC_UMBRELLA_MACHINE_NAME.equals(destination)){
                            mq4UmbrellaMachineNotify.receive(msg);
                        }else {
                            LOGGER.error("未找到消费者：{}", msg);
                        }
                    } catch (Exception e) {
                        LOGGER.error("消费：{}", msg,e);
                    }
                }
                // 如果是Map消息
                else if (message instanceof ActiveMQMapMessage) {
                    ActiveMQMapMessage mm = (ActiveMQMapMessage) message;
                    try {
                        LOGGER.info("from" + mm.getJMSDestination() + " get textMessage：\t" + mm.getString("msgId"));
                    } catch (Exception e) {
                        LOGGER.error("Map消费：{}", mm,e);
                    }
                }
                // 如果是Object消息
                else if (message instanceof ActiveMQObjectMessage) {
                    ActiveMQObjectMessage om = (ActiveMQObjectMessage) message;
                    //Object(om)
                }
                // 如果是Stream消息
                else if (message instanceof ActiveMQStreamMessage) {
                    ActiveMQStreamMessage sm = (ActiveMQStreamMessage) message;
                    try {
                        LOGGER.info(sm.readString());
                    } catch (Exception e) {
                        LOGGER.error("Stream消费：{}", sm,e);
                    }
                }
            }
        });
    }

}
